package com.abyss.util;

import com.abyss.config.QuotConfig;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Date 2020/9/15
 */
public class HbaseUtil {

    /**
     * 开发步骤：
     * 1.静态代码块获取连接对象
     * 2.获取表
     * 3.插入单列数据
     * 4.插入多列数据
     * 5.根据rowkey查询数据
     * 6.根据rowkey删除数据
     * 7.批量数据插入
     */
    static Connection connection;

    //获取连接
    static {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", QuotConfig.config.getProperty("hbase.zookeeper.quorum"));
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //2.获取表
    public static Table getTable(String tableName){

        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return table;
    }

    //3.插入单列数据
    public static void putDataByRowkey(String tableName,String family,String colName,String colVal,String rowkey){

        Table table = getTable(tableName);
        try {
            Put put = new Put(rowkey.getBytes());
            put.addColumn(family.getBytes(),colName.getBytes(),colVal.getBytes());
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //4.插入多列数据
    public static void putMapDataByRowkey(String tableName, String family , Map<String,Object> map, String rowkey){

        Table table = getTable(tableName);
        try {
            Put put = new Put(rowkey.getBytes());
            for (String key : map.keySet()) {
                //map得key就是列名，value就是列值
                put.addColumn(family.getBytes(),key.getBytes(),map.get(key).toString().getBytes());
            }
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //5.根据rowkey查询数据
    public static String queryByRowkey(String tableName,String family,String colName,String rowkey){

        Table table = getTable(tableName);
        String str = null;
        try {
            Get get = new Get(rowkey.getBytes());
            Result result = table.get(get);
            byte[] value = result.getValue(family.getBytes(), colName.getBytes());
            str = new String(value);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return str;
    }

    //6.根据rowkey删除数据
    public static void delByRowkey(String tableName,String family,String rowkey){

        Table table = getTable(tableName);
        try {
            Delete delete = new Delete(rowkey.getBytes());
            delete.addFamily(family.getBytes());
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //7.批量数据插入
    public static void putList(String tableName, List<Put> list){

        Table table = getTable(tableName);
        try {
            table.put(list);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //方法测试
    public static void main(String[] args) {

        putDataByRowkey("test","f1","age","20","16");
        HashMap<String, Object> map = new HashMap<>();
        map.put("age",21);
        map.put("name","小李");
        putMapDataByRowkey("test","f1",map,"17");
        String str = queryByRowkey("test", "f1", "age", "16");
        System.out.println(str);
        delByRowkey("test","f1","12");
        List<Put> list = new ArrayList<>();
        Put put = new Put("18".getBytes());
        put.addColumn("f1".getBytes(),"age".getBytes(),"22".getBytes());
        Put put1 = new Put("19".getBytes());
        put1.addColumn("f1".getBytes(),"age".getBytes(),"23".getBytes());
        list.add(put);
        list.add(put1);
        putList("test",list);

    }



}
